package com.shujia.flink.source

import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.connector.file.src.FileSource
import org.apache.flink.core.fs.Path
import org.apache.flink.formats.csv.CsvReaderFormat
import org.apache.flink.streaming.api.scala._

import java.time.Duration

object Demo3StreamFileSource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //定义数据的格式，数据结构
    val csvFormat: CsvReaderFormat[Student] = CsvReaderFormat
      .forPojo(classOf[Student])

    /**
     * 新版的FileSource  ,可以是有界流，也可以说无界流
     */
    val studentSource: FileSource[Student] = FileSource
      //指定读取数据的格式以及路径
      .forRecordStreamFormat(csvFormat, new Path("flink/data/students"))
      //指定每个多久时间扫描一次目录
      .monitorContinuously(Duration.ofSeconds(1))
      .build()


    val studentDS: DataStream[Student] = env
      .fromSource(studentSource, WatermarkStrategy.noWatermarks(), "studentSource")


    studentDS.print()

    env.execute()

  }

}

